package zju.dgm.service.distsqlservice;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import zju.dgm.dto.DataFrame;
import zju.dgm.dto.ServerInfo;
import zju.dgm.service.ServiceEntry;
import zju.dgm.service.dataservice.DataService;
import zju.dgm.service.dataservice.DataUtilProvider;
import zju.dgm.service.election.ElectionService;
import zju.dgm.service.loadbalance.LoadBalanceUtilProvider;
import zju.dgm.service.name.NameService;

import javax.annotation.Resource;
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

import zju.dgm.service.region.RegionUtilProvider;
import zju.dgm.sqlcfg.JdbcHandle;
import zju.dgm.util.SyncUtil;

@Component
public class DistSqlService extends ServiceEntry implements DistSqlUtilProvider {
    private static final Logger logger = LoggerFactory.getLogger(DistSqlService.class);

    public static CountDownLatch finLatch = new CountDownLatch(1);

    @Resource
    protected DataUtilProvider dataService;                 // 数据传输服务

    @Resource
    private RegionUtilProvider regionService;               // 副本管理服务

    @Resource
    private LoadBalanceUtilProvider loadBalanceService;     // 负载均衡服务

    @Resource
    private JdbcHandle jdbcHandle;      // 使用封装的Jdbc句柄，交由SpringBoot管理

    private ConcurrentHashMap<String, ArrayList<String>> logList;    // 日志用于回滚记录，可能占用空间

    private final ReturnFilter filter = new ReturnFilter();

    private final Object lock = new Object();

    private DataFrame getDataFrame;

    private class ReqListener extends Thread {
        @Override
        public void run() {
            logger.info("分布式查询开始监听消息");
            while (true){
                try {
                    DataFrame req;
                    logger.info("等待接收请求中...");
                    req = dataService.blockingGetOneDF(new ListenerFilter(), Integer.MAX_VALUE);
                    logger.info("接收到请求{}", req);
                    synchronized (lock){
                        getDataFrame = req;
                        lock.notifyAll();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private class Executor extends Thread{
        @Override
        public void run() {
            while (true){
                DataFrame req;
                try {
                    synchronized (lock){
                        lock.wait();
                        req = getDataFrame;
                    }
                    logger.info("开始执行线程");
                    DataFrame result;
                    ServerInfo sender = SyncUtil.serverInfoMap.get(req.getSender());

                    switch (req.getType()){
                        case DataFrame.SERVER_REQUEST:
                            result = readCall(req);
                            if (result == null){
                                result = new DataFrame(SyncUtil.serverName, DataFrame.ERR, null);
                            }
                            dataService.sendData(sender, result);
                            break;
                        case DataFrame.MASTER_WRITE_REQ:
//                      master转发给本server执行的写请求，调用本地写即可
                            String res = nonMasterWriteCall(req);
                            result = sealMessage(SyncUtil.serverName, getTable(req), RET, res, DataFrame.SERVER_RESPOND);
                            dataService.sendData(sender, result);
                            break;
                        case DataFrame.SERVER_TRANSFER_WRITE:
                            result = writeCall(req);
                            if (result == null){
                                result = new DataFrame(SyncUtil.serverName, DataFrame.ERR, null);
                            }
                            dataService.sendData(sender, result);
                            break;
                        case DataFrame.UNDO_REQUEST:
                            String rres = undoCall(getTable(req));
                            result = sealMessage(SyncUtil.serverName, getTable(req), RET, rres, DataFrame.SERVER_RESPOND);
                            dataService.sendData(sender, result);
                            break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private final ReqListener listener = new ReqListener();

    private final Executor executor = new Executor();

    @Override
    public String getRootPath() {
        return null;
    }

    @Override
    public boolean needRootPath() {
        return false;
    }

    @Override
    protected void registerListener() {
    }

    @Override
    protected CountDownLatch[] getPrerequisites() {
//        需要完成所依赖的服务
        return new CountDownLatch[]{NameService.finLatch, ElectionService.finLatch, DataService.finLatch};
    }

    @Override
    protected void start(){
        logger.info("分布式查询服务启动");
        logList = new ConcurrentHashMap<>();
        listener.start();
        executor.start();
        logger.info("本地ip:{}, MasterName:{}", SyncUtil.serverInfo.getIp(), SyncUtil.masterName);
        finLatch.countDown();
        logger.info("分布式查询服务完成");
    }

    @Override
    public String distSql(String table, int type, String sql) {
        DataFrame df = sealMessage(SyncUtil.serverName, table, type, sql, DataFrame.CLIENT_REQUEST);

        DataFrame ret;
        String res = null;
        try {
            switch (type){
                case SELECT:
                    ret = readCall(df);
                    res = unsealMessage(ret);
                    break;
                case CREATE:
                case DELETE:
                case UPDATE:
                    ret = writeCall(df);
                    res = unsealMessage(ret);
                    break;
            }
        } catch (Exception e) {
            e.printStackTrace();
            res = e.getMessage();
        }

        return res;
    }

//    @Override
    public DataFrame readCall(DataFrame request) throws Exception {
//        readCall类型检查，只支持SELECT
        int type = getSqlType(request);
        if (type != SELECT){
            throw new Exception("Read Call should use type: SELECT!");
        }

        String table = getTable(request);

//        本地表本地查询
        if (checkTableOwnership(table)){
//            MiniSQL查询得到JSON字符串，暂时不做解析
            String result = jdbcHandle.sql(getContent(request));
            loadBalanceService.touch(table);

//            检验返回DataFrame类型
            int returnType = -1;
            if (request.getType()==DataFrame.CLIENT_REQUEST){
                returnType = DataFrame.CLIENT_RESPOND;
            }else if(request.getType()==DataFrame.SERVER_REQUEST){
                returnType = DataFrame.SERVER_RESPOND;
            }

//          返回封装数据包
            return sealMessage(SyncUtil.serverName, table, RET, result, returnType);
        }
//        转发查询
        else{
//            通过负载均衡接口得到应当查询的表位置
            ServerInfo target = loadBalanceService.getRequestingServer(table);
//            使用DataService将读消息转发至对应的server, 设置类型
            DataFrame transfer = new DataFrame(SyncUtil.serverInfo.getIp(), DataFrame.SERVER_REQUEST, request.getData());
//            假设有返回开始解析返回数据
            DataFrame result = null;
            try {
                dataService.sendData(target, transfer);

                result = dataService.blockingGetOneDF(filter, 5000);

            }catch (Exception e){
                e.printStackTrace();
            }

//            获得值为null的DataFrame，出现异常
            if(result==null || result.getType()==DataFrame.ERR) throw new Exception("ERROR on Distribute sql.");

            result.setSender(SyncUtil.serverName);
            result.setType(DataFrame.CLIENT_RESPOND);

            return result;
        }
    }

//    接到写请求时调用，判断自己是否为master，是则执行，否则转发
    public DataFrame writeCall(DataFrame request) throws Exception {
//        Sql类型检查
        int type = getSqlType(request);
        if (type != CREATE && type!= DELETE && type != UPDATE){
            throw new Exception("Wrong type for Write Call!");
        }

        if (!SyncUtil.serverName.equals(SyncUtil.masterName)){
//            自己不是Master节点，所有写请求都应转发至Master
            DataFrame transfer = new DataFrame(SyncUtil.serverName, DataFrame.SERVER_TRANSFER_WRITE, request.getData());
            DataFrame result;
            logger.info("转发写请求至Master");
            dataService.sendData(SyncUtil.serverInfoMap.get(SyncUtil.masterName), transfer);
            result = dataService.blockingGetOneDF(filter, 5000);

            if (result==null || result.getType()==DataFrame.ERR) {
                logger.error("错误的result为: {}", result);
                throw new Exception("ERROR on Distribute write.");
            }

            result.setSender(SyncUtil.serverName);
            result.setType(DataFrame.CLIENT_RESPOND);

            return result;
        }

//        Master节点de写逻辑
        logger.info("Master节点处理写请求");
        DataFrame result = null;
        switch (type){
            case CREATE:
                result = createCall(request);
                break;
            case DELETE:
                result = deleteCall(request);
                break;
            case UPDATE:
                result = updateCall(request);
                break;
        }
        if (result == null) throw new Exception("Master write operation failed.");
        return result;
    }

    private DataFrame createCall(DataFrame req) throws Exception{
        List<ServerInfo> targets;
        DataFrame result = null;
//        获取创建位置, 得到ServerInfo数组
        targets = regionService.addNewRegion(getTable(req));
        boolean[] doneRecord = new boolean[targets.size()];
        boolean needReDo = false;
        for (int i=0; i<targets.size(); i++){
//            本地建表
            if (targets.get(i).getName().equals(SyncUtil.serverName)){
                String sRes = nonMasterWriteCall(req);
                if (sRes != null){
                    result = sealMessage(SyncUtil.serverName, getTable(req), RET, sRes, DataFrame.SERVER_RESPOND);
                    doneRecord[i] = true;
                    continue;
                }else{
                    needReDo = true;
                    break;
                }
            }

            DataFrame transfer = new DataFrame(SyncUtil.serverName, DataFrame.MASTER_WRITE_REQ, req.getData());
            logger.info("Master转发写请求至{}", targets.get(i).getName());
            dataService.sendData(targets.get(i), transfer);
            result = dataService.blockingGetOneDF(filter, 5000);

            if (result!=null&&result.getType()!=DataFrame.ERR) {
                logger.info("收到执行结果回复为: {}", getContent(result));
                doneRecord[i] = true;
            }
            else{
                logger.warn("{}执行写请求未成功", targets.get(i).getName());
                needReDo = true;
                break;
            }
        }
        if (needReDo){
            logger.info("开始执行回滚逻辑");
            for (int i=0; i<targets.size(); i++){
//                本地回滚
                if (doneRecord[i]&&targets.get(i).getName().equals(SyncUtil.serverName)){
                    String res = undoCall(getTable(req));
                    result = sealMessage(SyncUtil.serverName, getTable(req), RET, res, DataFrame.SERVER_RESPOND);
                }
                if (doneRecord[i]){
//                            回滚
                    DataFrame df = new DataFrame(SyncUtil.serverName, DataFrame.UNDO_REQUEST, null);
                    logger.info("向{}发送回滚请求", targets.get(i).getName());
                    dataService.sendData(targets.get(i), df);
                    result = dataService.blockingGetOneDF(filter, 5000);
                }
                logger.info("回滚结果{}", result);
            }
            return null;
        }

//                不需要回滚说明都创建完毕，通知regionServer更新map
        logger.info("Master执行写成功,result为{}",result);
        regionService.addConfirm(getTable(req), targets);

//                最后返回一个
        return result;
    }

    private DataFrame deleteCall(DataFrame req) throws Exception{
//      先获取table
        List<ServerInfo> targets = regionService.getRegionMap().get(getTable(req));
//      直接删信息，然后通知对应Server删除
        regionService.deleteRegion(getTable(req));
        boolean[] doneRecord = new boolean[targets.size()];
        DataFrame result = null;

        for (int i=0; i<targets.size(); i++){
//            Master本地删
            if (targets.get(i).getName().equals(SyncUtil.serverName)){
                String res = nonMasterWriteCall(req);
                if (res!=null){
                    doneRecord[i] = true;
                    result = sealMessage(SyncUtil.serverName, getTable(req), RET, res, DataFrame.SERVER_RESPOND);
                }
                continue;
            }

            DataFrame transfer = new DataFrame(SyncUtil.serverName, DataFrame.MASTER_WRITE_REQ, req.getData());

            dataService.sendData(targets.get(i), transfer);
            result = dataService.blockingGetOneDF(filter, 5000);

            if (result!=null&&result.getType()!=-1){
                doneRecord[i] = true;
            }
        }

//      失败的重做一遍尽力保证一致性
        for (int i=0; i<targets.size(); i++){
            if ((!doneRecord[i]) && targets.get(i).getName().equals(SyncUtil.serverName)){
                String res = nonMasterWriteCall(req);
                result = sealMessage(SyncUtil.serverName, getTable(req), RET, res, DataFrame.SERVER_RESPOND);
            }
            if (!doneRecord[i]){
                DataFrame transfer = new DataFrame(SyncUtil.serverName, DataFrame.MASTER_WRITE_REQ, req.getData());
                dataService.sendData(targets.get(i), transfer);
                result = dataService.blockingGetOneDF(filter, 5000);
            }
        }

        return result;
    }

    private DataFrame updateCall(DataFrame req) throws Exception {
//      逻辑和创建基本一致，仅不涉及RegionServer的map更新
//      获取更新位置
        List<ServerInfo> targets = regionService.getRegionMap().get(getTable(req));
        boolean[] doneRecord = new boolean[targets.size()];
        boolean needReDo = false;
        DataFrame result = null;
        for (int i=0; i<targets.size(); i++){
//            Master本地写
            if (targets.get(i).getName().equals(SyncUtil.serverName)){
                String res = nonMasterWriteCall(req);
                if (res!=null){
                    doneRecord[i] = true;
                    result = sealMessage(SyncUtil.serverName, getTable(req), RET, res, DataFrame.SERVER_RESPOND);
                }
                else{
                    needReDo = true;
                    break;
                }
                continue;
            }

            DataFrame transfer = new DataFrame(SyncUtil.serverName, DataFrame.MASTER_WRITE_REQ, req.getData());
            //根据返回值判断是否写入完成
            dataService.sendData(targets.get(i), transfer);
            result = dataService.blockingGetOneDF(filter, 5000);

            if (result!=null && result.getType()!=DataFrame.ERR) doneRecord[i] = true;
            else{
                needReDo = true;
                break;
            }
        }
        if (needReDo){
            for (int i=0; i<targets.size(); i++){
                if (targets.get(i).getName().equals(SyncUtil.serverName)&&doneRecord[i]){
                    undoCall(getTable(req));
                }
                else if (doneRecord[i]){
//                            回滚
                    DataFrame df = new DataFrame(SyncUtil.serverName, DataFrame.UNDO_REQUEST, null);
                    dataService.sendData(targets.get(i), df);
                }
            }
            return null;
        }
        return result;
    }

    @Override
    public DataFrame getLog(String table) throws IOException {
        ArrayList<String> operationLog = jdbcHandle.getOperationLog(table);

        ByteArrayOutputStream byteArrayStream = new ByteArrayOutputStream();
//        序列化为byte数组作为DataFrame的内容
        ObjectOutputStream out = new ObjectOutputStream(byteArrayStream);
//        将表名与日志一同封装
        out.writeObject(table);
        out.writeObject(operationLog);
        out.close();

        return new DataFrame(SyncUtil.serverName, DataFrame.LOG_INFO, byteArrayStream.toByteArray());
    }

    @Override
    public DataFrame reDoWithLog(DataFrame log) {
        byte[] logByte = log.getData();

        ByteArrayInputStream byteArrayStream = new ByteArrayInputStream(logByte);
        String table;
        ArrayList<String> readLog;
        try {
            ObjectInputStream in = new ObjectInputStream(byteArrayStream);
            table = (String) in.readObject();
            readLog = (ArrayList<String>) in.readObject();
            String res = jdbcHandle.redoOperationLog(table, readLog);
            return new DataFrame(SyncUtil.serverName, DataFrame.REDO_RESPOND, res.getBytes(StandardCharsets.UTF_8));
        } catch (IOException | ClassNotFoundException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 非Master逻辑，收到MasterCRT请求后执行的具体create操作
     * @param req 请求数据包
     */
    private String nonMasterWriteCall(DataFrame req){
        String res;
        String table = getTable(req);
        if (table==null) return "Error: No table name.";
        try {
//            写入前获取其日志
            ArrayList<String> operationLog = jdbcHandle.getOperationLog(table);
            logList.put(table, operationLog);
        } catch (IOException e) {
//            建表时没有原有日志，放入空list
            logList.put(table, new ArrayList<>());
        }

        try {
//            执行写入
            logger.info("本地开始执行写请求");
            res = jdbcHandle.sql(getContent(req));
        } catch (IOException e) {
            logger.warn("本地写请求失败");
            e.printStackTrace();
            return e.getMessage();
        }
        logger.info("完成本地写请求");
        return res;
    }

    /**
     * 回滚操作
     * @param table 需要回滚的表名
     * @return 回滚结果
     */
    private String undoCall(String table){
        try {
            String res = jdbcHandle.reBuildByLog(table, logList.get(table));
            logList.remove(table);
            return res;
        } catch (IOException e) {
            e.printStackTrace();
            return e.getMessage();
        }
    }

    /**
     * 确定表是否在本地Server上
     * @param table 表名
     * @return 存在返回true 不存在返回false
     * @throws Exception regionService抛出的异常
     */
    private boolean checkTableOwnership(String table) throws Exception{
//        查找本服务器存在的table名来确定归属关系
        Map<String, List<ServerInfo>> regionMap = regionService.getRegionMap();
        if (regionMap==null) throw new Exception("Error on getting region Map.");

        List<ServerInfo> regionList = regionMap.get(table);
        if (regionList==null) throw new Exception("Table does not exist.");
//        检查List中是否有本地Server，有则返回true
        return regionList.contains(SyncUtil.serverInfo);
    }

    public DataFrame sealMessage(String senderName, String table, int sqlType, String sql, int transferType) {

        ByteArrayOutputStream byteArrayStream = new ByteArrayOutputStream();
//        序列化为byte数组作为DataFrame的内容
        ObjectOutputStream out;
        try {
            out = new ObjectOutputStream(byteArrayStream);
            out.writeObject(table);
            out.writeObject(sqlType);
            out.writeObject(sql);
            out.close();
        } catch (IOException e) {
            e.printStackTrace();
            return new DataFrame(senderName, -1, null);
        }

        return new DataFrame(senderName, transferType, byteArrayStream.toByteArray());
    }
    
    public String unsealMessage(DataFrame df) {
//        ip@@table@@type@@result
        return getContent(df);
    }

    private String getTable(DataFrame df){
        String[] res = unSealContent(df);
        if (res != null){
            return res[0];
        }
        return null;
    }

    private int getSqlType(DataFrame df){
        String[] res = unSealContent(df);
        if (res != null){
            return Integer.decode(res[1]);
        }
        return ERR;
    }

    private String getContent(DataFrame df){
        String[] res = unSealContent(df);
        if (res != null){
            return res[2];
        }
        return null;
    }

    private String[] unSealContent(DataFrame df){
        String[] res = new String[3];
        byte[] byteArr = df.getData();
        ByteArrayInputStream byteArrayStream = new ByteArrayInputStream(byteArr);
        try {
            ObjectInputStream in = new ObjectInputStream(byteArrayStream);
            res[0] = (String) in.readObject();
            res[1] = String.valueOf((int) in.readObject());
            res[2] = (String) in.readObject();
            in.close();
            return res;
        } catch (IOException | ClassNotFoundException e) {
            e.printStackTrace();
        }
        return null;
    }
}